package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.bean.TypeCodeConstant;
import com.atguigu.upp.service.CKDBService;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.UPPUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/11/12
 *
 *  ①确定从当天的哪个宽表中查询数据
 *  ②写入之前，保证幂等性，先删除今天要写入的分区
 *  ③执行写入
 *          把宽表表的TagCode按照 tagValue的类型分为4类
 *                  String | deciaml | Uint64 | Datetime
 *               每一类写入到对应的表中
 *
 */
public class WrIteBitmapToCKApp
{
    public static void main(String[] args) throws IOException {

        //接收参数
        String taskId = args[0];
        String doDate = args[1];

        //准备编程环境
        SqlSessionFactory sqlSessionFactory1 = UPPUtil.getSqlSessionFactoryByConfig("config.xml");
        SqlSessionFactory sqlSessionFactory2 = UPPUtil.getSqlSessionFactoryByConfig("ckconfig.xml");
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSessionFactory1.openSession());
        CKDBService ckdbService = new CKDBService(sqlSessionFactory2.openSession());

        //怎么知道今天的宽表中有哪些标签?
        //宽表由当天启用的Task 计算的标签合并得到，只需要知道今天要计算的Task对应的tagInfo即可
        List<TagInfo> tags = mysqlDBService.getTagInfoTodayExecute();

        //分类
        List<TagInfo> stringTags = new ArrayList<>();
        List<TagInfo> decimalTags = new ArrayList<>();
        List<TagInfo> bigintTags = new ArrayList<>();
        List<TagInfo> dateTags = new ArrayList<>();

        for (TagInfo tag : tags) {
            if (TypeCodeConstant.TAG_VALUE_TYPE_STRING.equals(tag.getTagValueType())){
                stringTags.add(tag);
            }else if (TypeCodeConstant.TAG_VALUE_TYPE_DECIMAL.equals(tag.getTagValueType())){
                decimalTags.add(tag);
            }else if (TypeCodeConstant.TAG_VALUE_TYPE_LONG.equals(tag.getTagValueType())){
                bigintTags.add(tag);
            }else {
                dateTags.add(tag);
            }
        }

        //
        //按照tagValueType，将每一个集合写入不同的bitamp表中
        insertBitmapToCk(stringTags,"user_tag_value_string",doDate,ckdbService);
        insertBitmapToCk(decimalTags,"user_tag_value_decimal",doDate,ckdbService);
        insertBitmapToCk(bigintTags,"user_tag_value_long",doDate,ckdbService);
        insertBitmapToCk(dateTags,"user_tag_value_date",doDate,ckdbService);

        //统一使用Spark的spark-submit功能把这个jar包提交到yarn，必须是一个Spark程序，一个Spark程序必须有SparkContext
        SparkSession sparkSession = UPPUtil.getSparkSession("MergeWideTableApp");


    }

    private static void insertBitmapToCk(List<TagInfo> tags,String table,String doDate,CKDBService ckdbService) {

        //获取当天宽表的名字
        String wideTable = PropertiesUtil.getProperty("upwideprefix") + doDate.replace("-","_");

        if (tags.size() > 0){

            String tagSql = tags.stream()
                                 .map(tag -> String.format(" ( '%s', %s ) ", tag.getTagCode(), tag.getTagCode()))
                                 .collect(Collectors.joining(","));

            //先删除当天已经写入的数据
            ckdbService.deleteBitMapData(table,doDate);
            //再写入
            ckdbService.insertBitmapSql(table,doDate,tagSql,wideTable);

        }

    }


}
